/*    file: URtpPsMediaCaster.cpp
 *    desc:
 * 
 * created: 2016-12-22
 *  author: chuanjiang.zh@qq.com
 * company: 
 */

#include "URtpPsMediaCaster.h"
#include "TFileUtil.h"
#include "CLog.h"
#include "H264NaluParser.h"
#include "TStringCast.h"
#include "RtpHeader.h"
#include <limits>
#include "AppConst.h"
#include "Ffmpeg.h"
#include "RtpConst.h"

static const uint8_t	PS_RTP_PAYLOAD = 96;

namespace av
{

URtpPsMediaCaster::URtpPsMediaCaster() :
	m_socket(),
	m_addr(),
	m_id(),
	m_delayCount(),
	m_videoLastPts(AV_NOPTS_VALUE),
	m_audioLastPts(AV_NOPTS_VALUE),
	m_seq(1),
	m_ssrc(0x10),
	m_payload(PS_RTP_PAYLOAD),
	m_initTime()
{
	m_buffer.ensure(1024 * 400);

	m_muxer.reset(new PSMuxer());

}

URtpPsMediaCaster::~URtpPsMediaCaster()
{
}

int URtpPsMediaCaster::open(const std::string& ip, uint16_t port, const MediaFormat& fmt)
{
	CLog::debug("open. ip:%s port:%d\n", ip.c_str(), port);

	CLog::debug("video fmt. codec:%d,width:%d,height:%d,profile:%d,clock:%d\n",
		fmt.m_codec, fmt.m_width, fmt.m_height, fmt.m_profile, fmt.m_clockRate);
	if (fmt.getVideoPropSize() > 0)
	{
		std::string videoProp = comn::StringCast::toHexString(fmt.getVideoProp(), fmt.getVideoPropSize());
		CLog::debug("video config. %s\n", videoProp.c_str());
	}

	CLog::debug("audio fmt. codec: %d, freq:%d, channels:%d, profile:%d, clock:%d\n",
		fmt.m_audioCodec, fmt.m_sampleRate, fmt.m_channels, fmt.m_audioProfile, fmt.m_audioRate);
	if (fmt.getAudioConfigSize() > 0)
	{
		std::string audioConfig = comn::StringCast::toHexString(fmt.getAudioConfig(), fmt.getAudioConfigSize());
		CLog::debug("audio config. %s\n", audioConfig.c_str());
	}

	close();

	if (!openSocket(ip, port))
	{
		return ENODEV;
	}

	m_format = fmt;

	bool result = m_muxer->setCodec(fmt.m_codec, fmt.m_audioCodec);
	int ret = result ? 0 : EBADF;

	//m_tcpSocket.open(SOCK_STREAM);
	//m_tcpSocket.connect("192.168.3.53", 5500);

	m_propDetector.setCodec(fmt.m_codec);
	m_naluJoiner.reset(createNaluJoiner(fmt.m_codec));

	start();

	return ret;
}

void URtpPsMediaCaster::close()
{
	if (isRunning())
	{
		CLog::debug("URtpPsMediaCaster::close.\n");

		stop();
	}

	// m_muxer->clear();

	closeSocket();

	m_delayCount = 0;

	//m_tcpSocket.close();

	m_seq = 0;
	m_initTime = 0;

	m_propDetector.clear();
}

bool URtpPsMediaCaster::isOpen()
{
	return isRunning();
}

void URtpPsMediaCaster::setID(int id)
{
	comn::AutoCritSec lock(m_cs);
	m_id = id;
}

int URtpPsMediaCaster::getID()
{
	return m_id;
}

bool URtpPsMediaCaster::addTarget(const std::string& ip, uint16_t port)
{
	CLog::debug("addTarget. ip:%s port:%d\n", ip.c_str(), port);

	NetAddress addr(ip, port);

	bool done = false;
	comn::AutoCritSec lock(m_cs);
	NetAddressArray::iterator it = std::find(m_targets.begin(), m_targets.end(), addr);
	if (it == m_targets.end())
	{
		m_targets.push_back(addr);

		done = true;
	}

	return done;
}

void URtpPsMediaCaster::removeTarget(const std::string& ip, uint16_t port)
{
	CLog::debug("removeTarget. ip:%s port:%d\n", ip.c_str(), port);

	NetAddress addr(ip, port);

	comn::AutoCritSec lock(m_cs);

	NetAddressArray::iterator it = std::remove(m_targets.begin(), m_targets.end(), addr);
	if (it != m_targets.end())
	{
		m_targets.erase(it);
	}
}

size_t URtpPsMediaCaster::getTargetCount()
{
	comn::AutoCritSec lock(m_cs);
	return m_targets.size();
}

void URtpPsMediaCaster::removeAllTarget()
{
	comn::AutoCritSec lock(m_cs);
	m_targets.clear();
}

struct PacketHeader
{
	int prefix;
	int type;
	int size;
	int64_t pts;
	int duration;
	int flags;
};

bool URtpPsMediaCaster::write(int type, uint8_t* data, int size, int64_t pts, int duration, int flags)
{
	comn::AutoCritSec lock(m_cs);

	if (type == MEDIA_TYPE_VIDEO)
	{
		if (m_format.m_clockRate > 0 && m_format.m_clockRate != 90000)
		{
			pts = pts * 90000 / m_format.m_clockRate;
		}

		if (m_videoLastPts > 0)
		{
			if (m_videoLastPts > pts)
			{
				CLog::warning("(%d) video pts is invalid. last:%lld, now:%lld\n", m_id, m_videoLastPts, pts);
			}
		}

		bool isKeyFrame = false;
		int naluType = 0;
		m_propDetector.detect(data, size, isKeyFrame, naluType);

		flags |= isKeyFrame ? MEDIA_FLAG_KEY : 0;

		MPacket inPkt;
		inPkt.type = MEDIA_TYPE_VIDEO;
		inPkt.data = data;
		inPkt.size = size;
		inPkt.pts = pts;
		inPkt.flags = flags;
		inPkt.duration = 0;

		MPacket outPkt;
		memset(&outPkt, 0, sizeof(outPkt));

		if (m_naluJoiner->write(inPkt, naluType, outPkt))
		{
			bool keyFrame = (outPkt.flags & MEDIA_FLAG_KEY) != 0;
			uint8_t* outData = 0;
			int outSize = 0;
			m_muxer->writeVideo(outPkt.data, outPkt.size, keyFrame,
				outPkt.pts, outData, outSize);
			if (outSize > 0)
			{
				writePacket(outData, outSize, outPkt.pts, outPkt.flags);
			}
		}
		else
		{
			// pass
		}

		m_videoLastPts = pts;

		//CLog::debug("write video. size:%d. pts:%lld, duration:%d.    elapse:%d\n",
		//				size, pts, duration, (int)elapse);

		return true;
	}
	else
	{
		if (m_format.m_audioRate > 0 && m_format.m_audioRate != 90000)
		{
			pts = pts * 90000 / m_format.m_audioRate;
		}

		m_audioLastPts = pts;

		uint8_t* outData = 0;
		int outSize = 0;
		m_muxer->writeAudio(data, size, pts, outData, outSize);
		if (outSize > 0)
		{
			writePacket(outData, outSize, pts, flags);
		}

		return true;
	}
}

bool URtpPsMediaCaster::write(av::MediaPacketPtr& pkt)
{
	size_t size = m_pktQueue.push(pkt);
	if (size > (size_t)AppConst::QUEUE_SIZE)
	{
		m_pktQueue.clear();
	}
	return true;
}

void URtpPsMediaCaster::onMediaFormat(const MediaFormat& fmt)
{
	/// pass
}

void URtpPsMediaCaster::onMediaPacket(MediaPacketPtr& pkt)
{
	// pass
}

void URtpPsMediaCaster::onMediaEvent(int event)
{
	// pass
}

int URtpPsMediaCaster::run()
{
	while (!m_canExit)
	{
		MediaPacketPtr pkt;
		m_pktQueue.pop(pkt, -1);
		if (pkt)
		{
			handleMediaPacket(pkt);
		}
		else
		{
			//handleCloseFile();
		}
	}
	return 0;
}

void URtpPsMediaCaster::doStop()
{
	m_pktQueue.cancelWait();
}

void URtpPsMediaCaster::writePacket(uint8_t* data, int size, int64_t pts, int flags)
{
	//static int s_count = 0;
	//s_count++;
	MediaPacket pkt;
	pkt.set(data, size);
	pkt.pts = pts;
	pkt.flags = flags;

	m_packager.slice(pkt, av::RtpConst::RTP_PACKET_SIZE, this);
}

void URtpPsMediaCaster::onSlicePacket(const RtpPacket& pkt)
{
	if (m_initTime == 0)
	{
		m_initTime = 1 + 3600 + (std::numeric_limits<uint32_t>::max() - pkt.ts);
	}

	comn::AutoCritSec lock(m_cs);

	m_buffer.clear();
	m_buffer.ensure(pkt.size + sizeof(RTP_FIXED_HEADER));

	RTP_FIXED_HEADER header;
	memset(&header, 0, sizeof(header));
	header.version = 2;
	header.marker = pkt.mark;

	header.timestamp = htonl(m_initTime + pkt.ts);

	header.ssrc = htonl(m_ssrc);
	header.seq_no = htons(m_seq++);
	header.payload = m_payload;

	m_buffer.write((char*)&header, sizeof(header));

	m_buffer.write(pkt.data, pkt.size);

	sendPacket(m_buffer.data(), m_buffer.size());
}

bool URtpPsMediaCaster::openSocket(const std::string& ip, int port)
{
	if (!m_socket.isOpen())
	{
		if (!m_socket.open(SOCK_DGRAM))
		{
			return false;
		}
	}

	bool done = false;
	comn::SockAddr localAddr(ip, port);
	int ret = m_socket.bind(localAddr);
	if (ret == 0)
	{
		m_socket.setSendBufferSize(1024 * 1024);
		done = true;
	}
	else
	{
		m_socket.close();
	}
	return done;
}

void URtpPsMediaCaster::closeSocket()
{
	m_socket.close();
}

void URtpPsMediaCaster::sendPacket(uint8_t* data, int size)
{
	for (size_t i = 0; i < m_targets.size(); ++i)
	{
		m_delayCount++;
		if (m_delayCount >= 5)
		{
			m_delayCount = 0;
			timedwait(1);
		}

		NetAddress& addr = m_targets[i];
		int ret = sendPacket(addr, data, size);
		if (ret != size)
		{
			CLog::warning("sendTo %s:%d failed ret %d. size:%d\n",
				addr.m_ip.c_str(), addr.m_port, ret, size);
		}

	}

	m_delayCount = 0;
}

int URtpPsMediaCaster::sendPacket(NetAddress& addr, uint8_t* data, int size)
{
	comn::SockAddr sockAddr(addr.m_ip, addr.m_port);
	int ret = m_socket.sendTo((char*)data, size, 0, sockAddr);
	return ret;
}

void URtpPsMediaCaster::timedwait(int ms)
{
	fd_set setEx;
	FD_ZERO(&setEx);
	FD_SET(m_socket.getHandle(), &setEx);

	timeval tv = { ms / 1000, (ms % 1000) * 1000 };

#ifdef WIN32
	select((int)m_socket.getHandle() + 1, NULL, &setEx, &setEx, &tv);
#else
	select((int)m_socket.getHandle() + 1, NULL, NULL, &setEx, &tv);
#endif //

	//int elapse = (int)(comn::ClockTime::getTime() - tmStart);
	//if (elapse > (ms + 2))
	//{
	//    CLog::debug("CUdpMediaCaster::timedwait. elapse: %d, raw:%d\n", elapse, ms);
	//}
}

int URtpPsMediaCaster::detectKeyFrame(uint8_t* data, int size)
{
	int flags = 0;
	if (H264NaluParser::startWithH264Code(data, size))
	{
		NaluPacket nalu;
		if (H264NaluParser::parseNalu(data, size, nalu))
		{
			if ((nalu.type == NaluPacket::NALU_SPS) ||
				(nalu.type == NaluPacket::NALU_PPS) ||
				(nalu.type == NaluPacket::NALU_SEI) ||
				(nalu.type == NaluPacket::NALU_IFRAME))
			{
				flags = AV_PKT_FLAG_KEY;
			}
		}
	}
	return flags;
}

void URtpPsMediaCaster::handleMediaPacket(MediaPacketPtr& pkt)
{
	write(pkt->type, pkt->data, pkt->size, pkt->pts, pkt->duration, pkt->flags);
}

void URtpPsMediaCaster::setPayload(uint8_t payload)
{
	m_payload = payload;
}

void URtpPsMediaCaster::setSsrc(uint32_t ssrc)
{
	m_ssrc = ssrc;
}

} //
